package com.yujiahao.bigdata.sql

import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{Encoder, _}

object SparkSQL01_UDAF_1 {
  def main(args: Array[String]): Unit = {
    //TODO 1、SparkSQL直接构建无法构建，包是私有的，使用伴生对象伴生类构建builder
    //SparkSQL对SparkCore进行了封装，所以环境也要发生变化
    val spark = SparkSession.builder
      .master("local")
      .appName("Word Count")
      .getOrCreate()
    //构建Dataframe
    val df: DataFrame = spark.read.json("data/user.json")

    //使用SQL的方式访问DataFrame
    df.createOrReplaceTempView("user")

    //自定义一个函数功能将文档里面的数据进行处理，向Spark中udf中进行注册
    spark.udf.register("haha", functions.udaf(new MyAvg))
    spark.sql("select haha(age) from user").show()

    //TODO 2、关闭资源
    spark.stop()
  }

  //TODO 自定义函数的步骤：
  // 1、自定义一个类继承Aggregator(这里要注意选择相应泛型的类）
  // 2、确定泛型：
  // 输入：这里考虑的是将年龄求平均值，且从文档中读取的一般来说我们要将泛型设置为Long
  // 缓冲：这里面是有两个参数进行逻辑计算的，所以也是两个（Long，Long）考虑到可视化不好，这里可以使用样例类将这两个进行封装
  // 输出：输出的结果应该也是一个Long型的
  // 3、重写4+2方法

  class MyAvg extends Aggregator[Long, AvgBuffer, Long] {

    //这个是对缓冲区的初始化，因此直接设置为0
    override def zero: AvgBuffer = {
      AvgBuffer(0L, 0L)
    }

    //这里就是将每一个输入的年龄和数量进行累加，这里的参数是缓冲中的年龄，和进来的年龄，所以累加，加一
    override def reduce(b: AvgBuffer, a: Long): AvgBuffer = {
      b.total += a
      b.cnt += 1
      b
    }

    //这里将每个缓冲区的数据进行两两合作
    override def merge(b1: AvgBuffer, b2: AvgBuffer): AvgBuffer = {
      b1.total += b2.total
      b1.cnt += b2.cnt
      b1
    }
   //完成计算
    override def finish(reduction: AvgBuffer): Long = {
      reduction.total / reduction.cnt
    }

    //这两个可以认为是序列化
    override def bufferEncoder: Encoder[AvgBuffer] = Encoders.product

    override def outputEncoder: Encoder[Long] = Encoders.scalaLong
  }

  //这里将缓冲中的数据使用样例类进行封装

  case class AvgBuffer(var total: Long, var cnt: Long)


}
